#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include "../lib/coreqmanager.hpp"
#include "../lib/coreutil.hpp"

using namespace pland;

QueueManager* QueueManager::instance_ = NULL;

QueueManager* QueueManager::Instance()
{
    if(instance_ == NULL)
        instance_ = new QueueManager;
    return instance_;
}


QueueManager::QueueManager()
{
    running_ = false;
    port_ = CORE_QUEUE_MANAGER_PORT;
}

void QueueManager::setPort(uint port)
{
    boost::mutex::scoped_lock lock(mutex_);
    port_ = port;
}

bool QueueManager::start()
{
    boost::mutex::scoped_lock lock(mutex_);
    if(running_ == true)
    {
        return false;
    }
    running_ = true;
    listener_ = boost::thread(&QueueManager::start_, this);
    return running_;
}

void QueueManager::start_()
{
    int sockfd = -1;
    int newsockfd = -1;
    int yes = 1;
    socklen_t clilen;
    struct sockaddr_in serv_addr, cli_addr;
    try
    {
        sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (sockfd < 0)
        {
            boost::mutex::scoped_lock lock(mutex_);
            utill_log(std::string(__FILE__) + " Error creating socket");
            running_ = false;
            return;
        }
        bzero((char *) &serv_addr, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;
        serv_addr.sin_addr.s_addr = INADDR_ANY;
        serv_addr.sin_port = htons(port_);
        if (bind(sockfd, (struct sockaddr *) &serv_addr,sizeof(serv_addr)) < 0)
        {
            boost::mutex::scoped_lock lock(mutex_);
            utill_log(std::string(__FILE__) + " Error binding to port " + utill_to_string(port_));
            close(sockfd);
            return;
        }
        if(setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) < 0)
        {
            boost::mutex::scoped_lock lock(mutex_);
            utill_log(std::string(__FILE__) + " Error setting socket options " + utill_to_string(port_));
            close(sockfd);
            return;
        }
        listen(sockfd,5);
        clilen = sizeof(cli_addr);
        utill_log(std::string(__FILE__) + " Listening on port " + utill_to_string(port_));
        while(1)
        {
            newsockfd = accept(sockfd,(struct sockaddr *) &cli_addr, &clilen);
            if(newsockfd < 0)
                continue;
            std::string queue_data = "";
            std::map<std::string, SharedDQueue>::iterator iter;
            {
                boost::mutex::scoped_lock lock(mutex_);
                for(iter = queues_.begin(); iter != queues_.end(); iter++)
                {
                    SharedDQueue q = iter->second;
                    queue_data += iter->first + ":";
                    queue_data += utill_to_string(q->size()) + "/";
                    queue_data += utill_to_string(q->max_size()) +"\n";
                }
            }
            std::string output = utill_to_string(queues_.size()) + "\n";
            output += queue_data;
            send(newsockfd, output.c_str(), output.length(), MSG_NOSIGNAL);
            close(newsockfd);
        }
    }
    catch(boost::thread_interrupted e)
    {
        running_ = false;
        if(sockfd > 0)
            close(sockfd);
        return;
    }
}

bool QueueManager::stop()
{
    boost::mutex::scoped_lock lock(mutex_);

    if(running_ == false)
    {
        return false;
    }
    listener_.interrupt();
    listener_.join();
    running_ = false;
    return running_;
}

bool QueueManager::addQueueu(std::string name, SharedDQueue q)
{
    boost::mutex::scoped_lock lock(mutex_);
    if(queues_.find(name) != queues_.end())
        return false;
    queues_.insert(std::make_pair(name, q));
    return true;
}


bool QueueManager::removeQueueu(std::string name)
{
    boost::mutex::scoped_lock lock(mutex_);
    if(name == "*")
    {
        queues_.clear();
        return true;
    }
    if(queues_.find(name) == queues_.end())
        return false;
    queues_.erase(name);
        return true;
}

void QueueManager::removeAll()
{
	boost::mutex::scoped_lock lock(mutex_);
	queues_.clear();
}
